package com.uw.uwrpc.core.consumer;

import com.alibaba.fastjson.JSON;
import com.uw.uwrpc.core.api.Filter;
import com.uw.uwrpc.core.api.RpcContext;
import com.uw.uwrpc.core.api.RpcRequest;
import com.uw.uwrpc.core.api.RpcResponse;
import com.uw.uwrpc.core.constant.ContextConstant;
import com.uw.uwrpc.core.exception.ErrorCode;
import com.uw.uwrpc.core.exception.RpcException;
import com.uw.uwrpc.core.util.SlidingTimeWindow;
import com.uw.uwrpc.core.meta.InstanceMeta;
import com.uw.uwrpc.core.util.MethodUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.ObjectUtils;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@Slf4j
public class UWInvocationHandler implements InvocationHandler {

    private final String service;
    private final Object object;

    private final RpcContext rpcContext;

    private final InstanceHolder instanceHolder;

    /**
     * 时间窗口统计调用异常
     */
    private final Map<String, SlidingTimeWindow> timeWindowMap = new HashMap<>();


    public UWInvocationHandler(String service, InstanceHolder instanceHolder, RpcContext rpcContext) {
        this.service = service;
        this.object = new Object();
        this.rpcContext = rpcContext;
        this.instanceHolder = instanceHolder;


        // 定时扫描将隔离的实例设置为半开模式。
        ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
        int initialDelay = Integer.parseInt(rpcContext.getParameters().get(ContextConstant.half_open_initial_delay));
        int delay = Integer.parseInt(rpcContext.getParameters().get(ContextConstant.half_open_delay));
        executorService.scheduleWithFixedDelay(instanceHolder::halfOpenInstance, initialDelay, delay, TimeUnit.MILLISECONDS);
    }

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) {
        // 避免调用服务提供者的本地方法，而不是暴露的接口。
        if (MethodUtil.checkObjectMethod(method)) {
            try {
                return method.invoke(object, args);
            } catch (IllegalAccessException | InvocationTargetException e) {
                throw new RpcException(ErrorCode.reflect_exception, e);
            }
        }

        RpcRequest rpcRequest = new RpcRequest();
        rpcRequest.setService(service);
        rpcRequest.setMethodSign(MethodUtil.getMethodSign(method));
        rpcRequest.setArgs(args);

        int retryTimes = Integer.parseInt(rpcContext.getParameters()
                .getOrDefault(ContextConstant.retry_time, "2"));

        for (int i = 0; i < retryTimes; i++) {
            try {
                return doInvoke(method, rpcRequest);
            } catch (Exception e) {
                if (e instanceof RpcException ex) {
                    // 如果是超时，进行重试调用。
                    if (ErrorCode.socket_timeout.equals(ex.getErrorCode())) {
                        log.debug("===> socket_timeout retry times: {}", i);
                        continue;
                    }
                }
                throw new RpcException(e);
            }
        }

        return null;
    }

    private Object doInvoke(Method method, RpcRequest rpcRequest) {
        RpcResponse<?> rpcResponse = new RpcResponse<>();
        // 拦截器前置处理
        for (Filter filter : rpcContext.getFilterList()) {
            if (!filter.before(rpcRequest, rpcResponse)) {
                return JSON.parseObject(JSON.toJSONString(rpcResponse.getData()), method.getReturnType());
            }
        }

        // 优先从半开实例列表中获取调用实例，如果没有，则从可用实例列表中获取。
        InstanceMeta providerInstance = instanceHolder.getHalfOpenInstance();
        if (ObjectUtils.isEmpty(providerInstance)) {
            // 负载均衡
            List<InstanceMeta> providerCluster = rpcContext.getRouter().route(instanceHolder.getActiveInstances());
            providerInstance = rpcContext.getLoadBalancer().choose(providerCluster);
        }


        // 发起 rpc 调用
        String url = providerInstance.toUrl();
        try {
            rpcResponse = rpcContext.getConsumerInvoker().invoke(rpcRequest, url);
        } catch (Exception e) {
            // 故障规则统计和故障隔离。
            SlidingTimeWindow timeWindow = timeWindowMap.computeIfAbsent(url, k -> new SlidingTimeWindow());
            timeWindow.record(System.currentTimeMillis());
            log.debug("instance {} in window with {}", url, timeWindow.getSum());

            int isolateThreshold = Integer.parseInt(rpcContext.getParameters()
                    .getOrDefault(ContextConstant.isolate_threshold, "3"));
            if (timeWindow.getSum() >= isolateThreshold) {
                instanceHolder.isolateInstance(providerInstance);
            }

            throw e;
        }

        // 拦截器后置处理
        for (Filter filter : rpcContext.getFilterList()) {
            filter.after(rpcRequest, rpcResponse);
        }

        // 拦截器返回前处理-异常情况
        if (!rpcResponse.isStatus()) {
            for (Filter filter : rpcContext.getFilterList()) {
                filter.afterCompletion(rpcRequest, rpcResponse, rpcResponse.getEx());
            }

            RpcException exception = rpcResponse.getEx();
            if(exception != null) {
                log.error("response error.", exception);
                throw exception;
            }
            return null;
        }

        // 拦截器返回前处理-正常情况
        for (Filter filter : rpcContext.getFilterList()) {
            filter.afterCompletion(rpcRequest, rpcResponse, null);
        }

        Object result = JSON.parseObject(JSON.toJSONString(rpcResponse.getData()), method.getReturnType());
        return result;
    }

}
